package com.guandata.cdc.model.source;

import com.google.common.collect.Maps;
import com.guandata.cdc.model.ConnectParam;
import com.guandata.cdc.model.DataTableType;
import lombok.NonNull;
import lombok.Synchronized;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.table.api.TableDescriptor;

import java.util.Map;

@Slf4j
public class ConnectOptionsLoader<T extends ConnectParam> {
    private final TableDescriptor.Builder builder;
    private final DataTableType dataTableType;
    private final Map<ConnectParam, String> options;

    private ConnectOptionsLoader(@NonNull DataTableType dataTableType, @NonNull TableDescriptor.Builder builder) {
        this.builder = builder;
        this.dataTableType = dataTableType;
        this.options = Maps.newHashMap();
    }

    public static <T extends ConnectParam> ConnectOptionsLoader<T> builder(
        DataTableType dataTableType,
        TableDescriptor.Builder builder
    ) {
        return new ConnectOptionsLoader<>(dataTableType, builder);
    }

    @Synchronized
    public ConnectOptionsLoader<T> add(@NonNull T key, @NonNull String value) {
        if(!key.getDataTableTypes().contains(dataTableType) || !key.getPreChecker().apply(options)) {
            log.warn("This key {} can't be adaptive for data table type: {}", key, dataTableType);
            return this;
        }
        builder.option(key.getParam(), value);
        options.put(key, value);
        return this;
    }
}
